package com.shujia.flink.tf;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import javax.xml.crypto.Data;

public class Demo5Reduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //lambda表达式
        DataStream<Tuple2<String, Integer>> kvDS = linesDS
                .map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT));
        //分组
        KeyedStream<Tuple2<String, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0);

        DataStream<Tuple2<String, Integer>> countDS = keyByDS
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    /**
                     * reduce方法每一条数据执行一次,每个key第一条数据不执行了
                     * @param kv1 代表的是之前的计算结果（状态）
                     * @param kv2 代表的是当前处理的数据
                     */
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> kv1, Tuple2<String, Integer> kv2) throws Exception {
                        System.out.println("kv1:" + kv1);
                        System.out.println("kv2:" + kv2);
                        //统计单词的数量
                        int count = kv1.f1 + kv2.f1;
                        return Tuple2.of(kv2.f0, count);
                    }
                });

        //countDS.print();

        //lambda表达式
        keyByDS.reduce((kv1, kv2) -> Tuple2.of(kv1.f0, kv1.f1 + kv2.f1)).print();


        /*
         * 聚合函数 sum max min
         */
        keyByDS.sum(1).print();


        env.execute();
    }
}
